package cluster

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"

	"gddgame.cc/galaxy/asteroid"
	"gddgame.cc/galaxy/utils"
	"gddgame.cc/galaxy/utils/serialize"

	"gddgame.cc/galaxy/utils/def"

	"gddgame.cc/galaxy/utils/uuid"
)

type BalanceService interface {
	Service
	Master(node Node)
}

// 自动负载均衡，其余节点进行代理
// 可以手动删除
type BalanceCenter struct {
	container    *container
	manages      map[string]*ServiceManager
	managerMutex sync.RWMutex
	serverMutex  sync.Mutex
	clientMutex  sync.Mutex

	// 总服务数
	count int32
	// single节点服务数
	waitNumber int32

	result sync.Pool

	heir string
}
type balance struct {
	def.Logger
	*asteroid.Room
	BalanceService

	ctx      context.Context
	cancelFn func()

	instance interface{}
	center   *BalanceCenter
	token    string

	replicate bool
	master    bool
	simple    bool
	init      bool
}

var (
	masterServices = make(map[string]BalanceService)
	//masterMutex = sync.RWMutex{}
)

func registerBalance(service BalanceService) {
	//masterMutex.Lock()
	//defer masterMutex.Unlock()
	masterServices[service.ID()] = service
}

func getBalanceService(module string) BalanceService {
	//masterMutex.RLock()
	//defer masterMutex.RUnlock()
	if mm, ok := masterServices[module]; ok {
		return mm
	}
	// 需要预先register
	panic(fmt.Errorf("service is miss:%s，need register first", module))
}

func init() {

}

func newBalance(container *container) *BalanceCenter {
	m := &BalanceCenter{
		container:    container,
		manages:      map[string]*ServiceManager{},
		managerMutex: sync.RWMutex{},
		serverMutex:  sync.Mutex{},
		clientMutex:  sync.Mutex{},

		count:      0,
		waitNumber: 0,
		heir:       "",
	}
	container.global.On("balance/load", func(message asteroid.Message) {
		// 如果在退出中，load返回false，不参与负载均衡
		if container.exit {
			message.Response(false)
		} else {
			message.Response(m.count)
		}
	})
	container.global.On("balance/heir", func(message asteroid.Message) {
		if container.exit {
			// 如果在退出中

			for _, manager := range m.manages {
				manager.Services(func(key, value interface{}) bool {
					balance := value.(*balance)
					if balance.master && balance.replicate {
						// 取消所有主从节点的权限
						balance.master = false
					}
					return true
				})
			}
			return
		}
		bag := serialize.GetBag()
		bag.SetList(message.List())
		target := bag.ReadString(0)
		heir := bag.ReadString(1)
		bag.Release()
		m.heir = heir
		node := container.region.CurrentNode()
		nextHeir := m.Balance(true)
		// 调整所有运行服务
		for _, manager := range m.manages {
			manager.Services(func(key, value interface{}) bool {
				balance := value.(*balance)
				if balance.master {
					if balance.replicate {
						if balance.deleteSlave(target) {
							balance.addSlave(m.heir)
						}
					}
				} else if node == m.heir {
					balance.master = true
					balance.Silent(false)
					balance.LocalEventAsync(MasterEvent, nil, message.GetFrom(), message.GetTarget())
					container.global.Other().
						List(balance.BalanceService.ID(), balance.token, container.region.CurrentNode()).
						Send("balance/switch")
					balance.addSlave(nextHeir)
				}
				return true
			})
			manager.Clients(func(key, value interface{}) bool {
				client := value.(*Client)
				if client.target == target {
					client.changeTarget(heir)
				}
				return true
			})
		}
	})
	container.global.On("balance/master", func(message asteroid.Message) {
		bag := serialize.GetBag()
		bag.SetList(message.List())
		module := bag.ReadString(0)
		token := bag.ReadString(1)
		bag.Release()
		identity := token
		manager := m.LoadManager(module)
		//fmt.Println("on master1")
		i, ok := manager.LoadService(identity)
		if !ok {
			//fmt.Println("on master2")
			m.serverMutex.Lock()
			i, ok := manager.LoadService(identity)
			if ok {
				m.serverMutex.Unlock()
				message.Response(i.(*balance).master)
				return
			}
			m.serverMutex.Unlock()
			//fmt.Println("on master3")
			message.Response(false)
		} else {
			message.Response(i.(*balance).master)
		}
	})
	container.global.On("balance/create", func(message asteroid.Message) {
		bag := serialize.GetBag()
		bag.SetList(message.List())
		module := bag.ReadString(0)
		token := bag.ReadString(1)
		master := bag.ReadBool(2)
		bag.Release()
		identity := token
		manager := m.LoadManager(module)
		_, ok := manager.LoadService(identity)
		if !ok {
			m.serverMutex.Lock()
			_, ok := manager.LoadService(identity)
			if !ok {
				m.createService(module, token, master)
				atomic.AddInt32(&m.count, 1)
			}
			m.serverMutex.Unlock()
		}
		message.Response(true)
	})
	container.global.On("balance/delete", func(message asteroid.Message) {
		bag := serialize.GetBag()
		bag.SetList(message.List())
		module := bag.ReadString(0)
		token := bag.ReadString(1)
		bag.Release()
		identity := token
		manager := m.LoadManager(module)
		client, ok := manager.LoadClient(identity)
		if ok {
			client.(*Client).close()
			manager.DeleteClient(identity)
		}
		i, ok := manager.services.Load(identity)
		if ok {
			i.(*balance).close()
			manager.DeleteService(identity)
			atomic.AddInt32(&m.count, -1)
		}
		message.Response(true)
	})
	container.global.On("balance/switch", func(message asteroid.Message) {
		bag := serialize.GetBag()
		bag.SetList(message.List())
		module := bag.ReadString(0)
		token := bag.ReadString(1)
		target := bag.ReadString(2)
		bag.Release()
		identity := token
		manager := m.LoadManager(module)
		client, ok := manager.LoadClient(identity)
		if ok {
			client.(*Client).target = target
		}
	})
	m.result.New = func() interface{} {
		return make(chan string)
	}

	return m
}

func (m *BalanceCenter) LoadManager(module string) *ServiceManager {
	m.managerMutex.RLock()
	manager, ok := m.manages[module]
	m.managerMutex.RUnlock()
	if !ok {
		m.managerMutex.Lock()
		manager, ok = m.manages[module]
		if !ok {
			manager = NewManager(getBalanceService(module))
			m.manages[module] = manager
		}
		m.managerMutex.Unlock()
	}
	return manager
}

func (m *BalanceCenter) createService(module string, token string, master bool) bool {
	identity := token
	tr, _ := m.container.region.OpenRoom(module + "/" + identity)
	manager := m.LoadManager(module)
	service := manager.service.(BalanceService)
	ctx, cancelFn := context.WithCancel(m.container.engine.ctx)
	balance := &balance{
		Room:           tr,
		BalanceService: service,
		center:         m,
		token:          token,
		master:         master,
		ctx:            ctx,
		cancelFn:       cancelFn,
		replicate:      false,
		simple:         false,
		init:           false,
	}
	balance.Logger = tr.LoggerWrap("Cluster", module)
	// 保证流程
	tr.On(JoinEvent, func(message asteroid.Message) {
		message.Response(true)
	})
	tr.On(LeaveEvent, func(message asteroid.Message) {
		message.Response(true)
	})
	tr.On(MasterEvent, func(message asteroid.Message) {
		service.Master(balance)
	})
	tr.On("sync", func(message asteroid.Message) {
		tr.SwitchTap(true)
		b, err := message.Payload()
		if err == nil {
			if err = balance.instance.(def.Serializer).Unmarshal(b); err != nil {
				tr.Error(err)
			}
		} else {
			tr.Error(err)
		}

		tr.SwitchTap(false)
	})
	if !master {
		balance.Silent(true)
	}
	manager.StoreService(identity, balance)
	//m.container.engine.event.Fire("balance/create", module, balance.token, balance.instance)
	if balance.IsSingle() {
		atomic.AddInt32(&m.waitNumber, 1)
	}
	service.Init(balance)
	if master {
		if balance.replicate && m.container.region.Count() > 1 {
			switch balance.instance.(type) {
			case def.Serializer:
			default:
				m.container.Error("Balance instance need def.Serializer of replicate feature ", balance.instance)
				return false
			}
			slave := m.Balance(true)
			balance.addSlave(slave)
		}
		balance.LocalEventAsync(MasterEvent, nil, tr.CurrentNode(), tr.CurrentNode())
	}
	balance.init = true
	return true
}

func (m *BalanceCenter) closeClient(client *Client, service Service) {
	service.CloseClient(client)
	manager := m.LoadManager(client.service.ID())
	manager.DeleteClient(client.identity)
}

func (m *BalanceCenter) recovery(target string) {
	// 退出节点作为leader也没有问题，继位者是通过balance求的的
	// leader作用是保证逻辑唯一
	leader := m.container.Cluster().Leader()
	if leader != m.container.region.CurrentNode() {
		// 等待leader发布继任者
		if m.heir != target && m.heir != "" {
			// 继位者依旧有效
			return
		}
	} else {
		m.heir = m.Balance(false)
		m.container.global.All().List(target, m.heir).Send("balance/heir")
	}
}

func (m *BalanceCenter) generateMaster(module string, id string) string {
	master := m.result.Get().(chan string)
	defer m.result.Put(master)
	m.container.global.All().Callback(func(success bool, data map[string]interface{}) {
		//fmt.Println("master", success, data)
		for a := range data {
			switch v := data[a].(type) {
			case bool:
				if v {
					master <- a
					return
				}
			default:
			}
		}
		master <- ""
	}).List(module, id).Send("balance/master")
	target := <-master
	//fmt.Println("generate1", target)
	if target == "" {
		target = m.Balance(false)
		m.container.global.Direct(target).CallbackTo(func(success bool, data interface{}) {
			master <- target
		}).List(module, id, true).Send("balance/create")
		target = <-master
		//fmt.Println("generate2", target)
	}
	//fmt.Println("generate3", target)
	return target
}

func (m *BalanceCenter) getMaster(module string, id string) string {
	master := m.result.Get().(chan string)
	defer m.result.Put(master)
	m.container.global.All().List(module, id).Callback(func(success bool, data map[string]interface{}) {
		for a := range data {
			switch data[a].(type) {
			case bool:
				if data[a].(bool) {
					master <- a
					return
				}
			default:
			}
		}
		master <- ""
	}).Send("balance/master")
	return <-master
}
func (m *BalanceCenter) LocalService(module string, token string) *Client {
	identity := token
	manager := m.LoadManager(module)
	_, ok := manager.LoadService(identity)
	if !ok {
		m.serverMutex.Lock()
		_, ok := manager.LoadService(identity)
		if !ok {
			m.createService(module, token, true)
			atomic.AddInt32(&m.count, 1)
		}
		m.serverMutex.Lock()
	}
	return m.GenerateClient(module, token)
}

func (m *BalanceCenter) CreateService(module string) string {
	target := m.Balance(false)
	token := uuid.RandomString()
	result := make(chan string)
	m.container.global.Direct(target).List(module, token, true).CallbackTo(func(success bool, data interface{}) {
		if success {
			result <- token
		} else {
			result <- ""
		}
	}).Send("balance/create")
	return <-result
}

func (m *BalanceCenter) GenerateClient(module string, token string) *Client {
	identity := token
	var master string
	var tr *asteroid.Room
	var client *Client
	manager := m.LoadManager(module)
	i, ok := manager.LoadClient(identity)
	if ok {
		client = i.(*Client)
		goto Return
	}
	m.clientMutex.Lock()
	i, ok = manager.LoadClient(identity)
	if ok {
		m.clientMutex.Unlock()
		client = i.(*Client)
		goto Return
	}
	//fmt.Println("master0", master, token)
	master = m.generateMaster(module, token)
	//fmt.Println("master1", master, token)
	tr, _ = m.container.region.OpenRoom(module + "/" + identity)
	//fmt.Println("master2", master, token)
	client = newClient(m.container, token, master, m, manager.service, tr)
	manager.StoreClient(identity, client)
	manager.service.InitClient(client)
	//fmt.Println("master3", master, token)
	m.clientMutex.Unlock()
Return:
	return client
}

func (m *BalanceCenter) GetClient(module string, token string) *Client {
	identity := token
	var master string
	var tr *asteroid.Room
	var client *Client
	manager := m.LoadManager(module)
	i, ok := manager.LoadClient(identity)
	if ok {
		client = i.(*Client)
		goto Return
	}
	m.clientMutex.Lock()
	i, ok = manager.LoadClient(identity)
	if ok {
		m.clientMutex.Unlock()
		client = i.(*Client)
		goto Return
	}

	master = m.getMaster(module, token)
	if master == "" {
		goto Return
	}
	tr, _ = m.container.region.OpenRoom(module + "/" + token)
	client = newClient(m.container, token, master, m, manager.service, tr)
	manager.StoreClient(identity, client)
	manager.service.InitClient(client)
	m.clientMutex.Unlock()
Return:
	return client
}

func (m *BalanceCenter) Balance(except bool) string {
	result := make(chan string)
	handler := func(success bool, data map[string]interface{}) {
		min := -1
		master := ""
		if success {
			for node, value := range data {
				switch v := value.(type) {
				case bool:
					continue
				default:
					count := utils.ToInt(v)
					if min == -1 || count < min {
						min = count
						master = node
					}
				}
			}
		}

		result <- master
	}
	if except {
		m.container.global.Other().Callback(handler).Send("balance/load")
	} else {
		m.container.global.All().Callback(handler).Send("balance/load")
	}
	return <-result
}

func (m *BalanceCenter) ServiceRange(service string, f func(node Node, master bool) bool) {
	manager := m.LoadManager(service)
	manager.Services(func(key, value interface{}) bool {
		balance := value.(*balance)
		return f(balance, balance.master)
	})
}

func (i *balance) addSlave(target string) {
	// 创建slave
	i.center.container.global.Direct(target).CallbackTo(func(success bool, data interface{}) {
		i.SwitchTap(true)
		// 数据拷贝
		i.Direct(target).List(i.instance.(def.Serializer).Marshal()).Send("sync")
		// 添加监听
		i.AddReplica(target)
		i.SwitchTap(false)
	}).List(i.BalanceService.ID(), i.token, false).Send("balance/create")
}

func (i *balance) deleteSlave(target string) bool {
	return i.DeleteReplica(target)
}

func (i *balance) Instance() interface{} {
	return i.instance
}
func (i *balance) SetInstance(instance interface{}) {
	i.instance = instance
}
func (i *balance) Container() Container {
	return i.center.container
}
func (i *balance) Context() context.Context {
	return i.ctx
}
func (i *balance) Token() string {
	return i.token
}
func (i *balance) SetConfig(key NodeConfig, value interface{}) {
	if i.init {
		i.center.container.Error(key + " init finish, not allow change config")
		return
	}
	switch key {
	case ThreadSafe:
		i.SetThread(value.(bool))
	case Replicate:
		// 对于非replicate服务，必须是限时时间内会关闭
		// 否则影响安全退出，等所有single服务结束才会完成退出
		i.replicate = value.(bool)
	case Simple:
		// 简单服务，不考虑安全退出，是非replicate的替代选项
		i.simple = value.(bool)
	case SaveLocal:
		// 定时存储到本地
	default:
		i.center.container.Error(key + " is not effect config")
	}
}

func (i *balance) close() {
	i.CloseInstance(i)
	i.cancelFn()
	//i.Debugf("balance close：%s/%s", i.BalanceService.ID(), i.token)
}

// single节点判断
func (i *balance) IsSingle() bool {
	return i.master && !i.replicate && !i.simple
}
func (i *balance) Close() error {
	i.center.container.global.All().List(i.BalanceService.ID(), i.token).Send("balance/delete")
	// 本地服务数减一
	if i.IsSingle() {
		atomic.AddInt32(&i.center.waitNumber, -1)
		i.center.container.onCloseService()
	}
	// 直接关闭服务，避免内存溢出
	i.close()
	manager := i.center.LoadManager(i.BalanceService.ID())
	manager.DeleteService(i.token)
	atomic.AddInt32(&i.center.count, -1)
	return nil
}

func (m *BalanceCenter) close() {
	m.managerMutex.RLock()
	for _, manager := range m.manages {
		manager.Services(func(key, value interface{}) bool {
			value.(*balance).close()
			return true
		})
		manager.Clients(func(key, value interface{}) bool {
			value.(*Client).close()
			return true
		})
	}
	m.managerMutex.RUnlock()
}
